package com.bw.gmall.realtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.UserRegisterBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.DorisMapFunction;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: DwsUserUserRegisterWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/3 18:47
 * @version: 1.0
 */

public class DwsUserUserRegisterWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserRegisterWindow().start(Constant.TOPIC_DWD_USER_REGISTER, Constant.DWS_USER_USER_REGISTER_WINDOW, 4, 10025);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        SingleOutputStreamOperator<JSONObject> getEtl = getCreateTime(dataStreamSource);
//        getEtl.print();
        SingleOutputStreamOperator<UserRegisterBean> resultStream = getResultStream(getEtl);
        resultStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_USER_USER_REGISTER_WINDOW));
    }

    private static SingleOutputStreamOperator<UserRegisterBean> getResultStream(SingleOutputStreamOperator<JSONObject> getEtl) {
        return getEtl.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(10)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {

                return jsonObject.getLong("create_time");
            }
        })).windowAll(TumblingEventTimeWindows.of(Time.seconds(5))).process(new ProcessAllWindowFunction<JSONObject, UserRegisterBean, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<JSONObject, UserRegisterBean, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<UserRegisterBean> collector) throws Exception {
                ArrayList<JSONObject> jsonObjects = ListUtil.toList(iterable);
                long registerCt = jsonObjects.stream().map(e -> e.getString("create_time")).count();
                String start = DateFormatUtil.tsToDateTime(context.window().getStart());
                String end = DateFormatUtil.tsToDateTime(context.window().getEnd());
                String curDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                collector.collect(UserRegisterBean.builder()
                        .stt(start).edt(end).curDate(curDate).registerCt(registerCt)
                        .build());
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getCreateTime(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String create_time = jsonObject.getString("create_time");
                if (create_time != null) {
                    collector.collect(jsonObject);
                }
            }
        });
    }
}
